package epoch.game.core.distr;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.logging.log4j.Logger;

import epoch.game.core.action.IAction;
import epoch.game.core.common.Loggers;
import epoch.game.core.config.Config;
import epoch.game.core.context.Context;
import epoch.game.core.distr.call.CallContext;
import epoch.game.core.distr.call.CallResult;
import epoch.game.core.distr.call.CallReturn;
import epoch.game.core.exception.SysException;
import epoch.game.core.function.FuncUtil;
import epoch.game.core.function.IFunc1;
import epoch.game.core.function.IFunc2;
import epoch.game.core.thread.ServerThread;
import epoch.game.core.timer.GameTimer;
import epoch.game.core.util.BeanUtil;

/**
 * 节点线程上下文数据
 * 
 * @author will
 * @date 2016年3月31日 下午6:46:28
 */
public class ThreadContext {

	private static Logger log = Loggers.CORE_LOGGER;
	// 默认回调超时时间
	public static final int DEFAULT_TIMEOUT=60000;
	//portid
	int id;
	//节点id
	int nodeId;
	// 执行线程
	ServerThread thread;
	// 服务节点
	Node node;	
	// 当前线程的Port实例
	private final static ThreadLocal<ThreadContext> localContext = new ThreadLocal<ThreadContext>();
	//调用结果调用
	public ConcurrentLinkedQueue<CallReturn> callReturns = new ConcurrentLinkedQueue<CallReturn>();
	//请求id序列号生成器
	private AtomicLong callIdGen = new AtomicLong(1L);
	//当前调用绑定的回调id[用于监听和执行逻辑时候的传递，传递注册监听时候赋值，在执行方法中一定要将其置为0]
	public long currentCallbackId=0;

	//回调函数及回调上下文参数
	public HashMap<Long,CallResult> callbackMap = new HashMap<>();
	//当前正在等待同步方法返回结果的callId
	private long waitSyncCallId=0;
//	//等待同步调用结果<syncCallId,timeOut> 
//	private LinkedHashMap<Long,Long> waitSyncCalls = new LinkedHashMap<>();
	//心跳定时器
	private ThreadHeartTimer heartTimer = new ThreadHeartTimer(this, 1000);
	//上次心跳时间
	private long lastPulseTime;
	
	public int getId() {		
		return id;
	}
	
	public Node getNode() {
		return node;
	}

	public ServerThread getThread() {
		return thread;
	}

	/**
	 * 获取当前线程的ThreadContext实例，仅仅当调用线程为ServerThread子类时才可获取到
	 * 
	 * @return
	 */
	@SuppressWarnings("unchecked")
	public static <T extends ThreadContext> T getCurrent() {
		if (Thread.currentThread() instanceof ServerThread) {
			return (T) localContext.get();
		} else {
			return null;
		}
	}
	
	/**
	 * 设置上下文，仅仅当调用线程为ServerThread子类时才生效，以防设置后资源不能被及时释放
	 * @param context
	 */
	public static void setCurrentContext(ThreadContext context){
		if (Thread.currentThread() instanceof ServerThread) {
			localContext.set(context);
		}
	}
	
	/**
	 * 此构造函数会自动new一个线程
	 * @param node
	 */
	public ThreadContext(Node node){
		this.node = node;
		this.id = this.node.applyThreadContextId();
		this.nodeId = node.getId();
		this.node.addThreadContext(this);
		this.thread = new ServerThread("nodethread-"+this.id, 0);
		this.thread.setContext(this);
		//注册心跳定时器
		heartTimer = new ThreadHeartTimer(this, 1000);
		this.thread.registerTimer(heartTimer);
	}
	
	/**
	 * 构造函数
	 * @param node
	 * @param serverThread
	 */
	public ThreadContext(Node node, ServerThread serverThread){
		this.node = node;
		this.id = this.node.applyThreadContextId();
		this.nodeId = node.getId();
		this.node.addThreadContext(this);
		this.thread = serverThread;
		this.thread.setContext(this);
		// 注册心跳定时器
		heartTimer = new ThreadHeartTimer(this, 1000);
		this.thread.registerTimer(heartTimer);
	}
	
	public void registerTimer(GameTimer timer) {
		this.thread.registerTimer(timer);
	}
	
	/**
	 * 线程启动时设置上下文，调用线程需要能处理回调才能
	 * @param context
	 */
	public void onThreadStart() {
		lastPulseTime = System.currentTimeMillis();
		this.node.addThreadContext(this);
		localContext.set(this);
	}
	
	/**
	 * 当线程停止的处理
	 */
	public void onThreadStop(){
		this.node.removeThreadContext(this);
		this.node = null;
		localContext.set(null);
	}
	
	/**
	 * 启动线程
	 */
	public void start(){
		if(thread!=null){
			thread.start();
		}
	}
	
	/**
	 * 停止线程
	 * 
	 * @param force
	 */
	public void stop(boolean force) {
		if (thread != null) {
			// 停止独立线程
			this.thread.stop(force);
		}
	}
	


	/**
	 * 添加任务到执行线程
	 * @param action
	 */
	public void addAction(IAction action){
		thread.addAction(action);
	}

	

	public long applyCallId(){
		return callIdGen.getAndIncrement();
	}

	/**
	 * 监听调用结果
	 * @param callback
	 */
	public static <R> void listenResult(IFunc1<R> callback){
		ThreadContext current = ThreadContext.getCurrent();
		if (current != null) {
			current.listen(callback,false);
		} else {
			Context.listen(callback, null, false);
		}
	}
	
	/**
	 * 监听调用结果
	 * @param callback
	 * @param handleTimeOut 是否处理超时，如果为true，注意callback方法的参数必须为Param对象，是否超时结果的获取 context.get("isTimeOut");
	 */
	public static <R> void listenResult(IFunc1<R> callback, boolean handleTimeOut){
		ThreadContext current = ThreadContext.getCurrent();
		if (current != null) {
			current.listen(callback, null, handleTimeOut);
		} else {
			Context.listen(callback, null, handleTimeOut);
		}
	}
	
	/**
	 * 监听调用结果
	 * @param callback
	 * @param handleTimeOut 是否处理超时，如果为true，注意callback方法的参数必须为Param对象，是否超时结果的获取 context.get("isTimeOut");
	 */
	public static <R> void listenResult(IFunc2<CallContext,R> callback, boolean handleTimeOut){
		ThreadContext current = ThreadContext.getCurrent();	
		if (current != null) {
			current.listen(callback, handleTimeOut);
		} else {
			CallContext context = handleTimeOut ? new CallContext() : null;
			Context.listen(callback,context, handleTimeOut);
		}
	}
	
	public static <R> void listenResult(IFunc2<CallContext,R> callback, CallContext context){
		ThreadContext current = ThreadContext.getCurrent();
		if (current != null) {
			current.listen(callback,context, false);
		} else {
			Context.listen(callback,context, false);
		}
	}
	
	/**
	 * 监听调用，带回调会话数据
	 * @param callback 回调方法
	 * @param context 缓存的传递数据
	 * @param handleTimeOut 是否处理超时，如果为true，注意callback方法的参数必须为Param对象，是否超时结果的获取 context.get("isTimeOut");
	 */
	public static <R> void listenResult(IFunc2<CallContext,R> callback, CallContext context, boolean handleTimeOut){
		ThreadContext current = ThreadContext.getCurrent();	
		if (current != null) {
			current.listen(callback,context,handleTimeOut);
		} else {
			Context.listen(callback,context, handleTimeOut);
		}
	}
	
	/**
	 * 增加回调监听，需要在执行操作之前进行监听，否则无效
	 * @param callback
	 * @param context
	 */
	public static void listenResult(Object callback, CallContext context) {
		ThreadContext current = ThreadContext.getCurrent();
		if (current != null) {
			current.listen(callback,context,false);
		} else {
			Context.listen(callback,context, false);
		}
	}
	
	/**
	 * 增加回调监听，需要在执行操作之前进行监听，否则无效
	 * @param callback
	 * @param context
	 * @param sync 是否同步等待
	 * @param handleTimeOut 是否处理超时
	 */
	public static void listenResult(Object callback, CallContext context, boolean handleTimeOut) {
		ThreadContext current = ThreadContext.getCurrent();
		if (current != null) {
			current.listen(callback,context,handleTimeOut);
		} else {
			Context.listen(callback,context, handleTimeOut);
		}
	}
	
	/**
	 * 不注册回调，同步等待结果
	 * 
	 * <pre>
	 * 与waitForResult()配对使用，eg:
	 * xx.listenResultSync()
	 * xx.doSomething()
	 * Result ret = xx.waitForResult();
	 * </pre>
	 */
//	public static void listenResultSync(){
//		ThreadContext current = ThreadContext.getCurrent();
//		current.listenSync();
//	}
	
	/**
	 * 阻塞当前线程直到返回结果(业务线程中慎用，如果使用不当会造成处理其他逻辑超时，可能对游戏造成严重后果！！！)
	 * 
	 * <pre>
	 * 必须与listenResultSync()配对使用，否则只会返回null。eg:
	 * xx.listenResultSync()
	 * xx.doSomething()
	 * Result ret = xx.waitForResult();
	 * </pre>
	 * @return
	 */
//	public static <T> T waitForResult(){
//		return ThreadContext.getCurrent().waitFor(DEFAULT_TIMEOUT);
//	}
	
	/**
	 * 阻塞等待远程调用返回的结果，此方法谨慎使用，如果使用不当会造成处理其他逻辑超时，可能对游戏造成严重后果！！！
	 * <pre>
	 * 必须与listenResultSync()配对使用，否则只会返回null。eg:
	 * xx.listenResultSync()
	 * xx.doSomething()
	 * Result ret = xx.waitForResult();
	 * </pre>
	 * @param timeout 超时时间（ms）
	 * @return
	 */
//	public static <T> T waitForResult(long timeout){
//		return ThreadContext.getCurrent().waitFor(timeout);
//	}
	
	/**
	 * 监听调用结果
	 * @param callback
	 * @param handleTimeOut 是否处理超时，如果为true，注意callback方法的参数必须为Param对象，是否超时结果的获取 context.get("isTimeOut");
	 */
	private <R> void listen(IFunc1<R> callback, boolean handleTimeOut){
		listen(callback,null, handleTimeOut);
	}
	
	/**
	 * 监听调用结果
	 * @param callback
	 * @param handleTimeOut 是否处理超时，如果为true，注意callback方法的参数必须为Param对象，是否超时结果的获取 context.get("isTimeOut");
	 */
	private <R> void listen(IFunc2<CallContext,R> callback, boolean handleTimeOut){
		if(handleTimeOut){
			listen(callback,new CallContext(),handleTimeOut);
		}else{
			listen(callback,null,handleTimeOut);
		}
	}
	
	/**
	 * 监听调用，带回调会话数据R
	 * @param callback 回调方法
	 * @param context 缓存的传递数据
	 * @param handleTimeOut 是否处理超时，如果为true，注意callback方法的参数必须为Param对象，是否超时结果的获取 context.get("isTimeOut");
	 */
	private <R> void listen(IFunc2<CallContext,R> callback, CallContext context, boolean handleTimeOut){
		listen((Object)callback,context,handleTimeOut);
	}
	
	/**
	 * 增加回调监听，需要在执行操作之前进行监听，否则无效
	 * @param callback
	 * @param context
	 * @param sync 是否同步等待
	 * @param handleTimeOut 是否处理超时
	 */
	private void listen(Object callback, CallContext context, boolean handleTimeOut) {
		CallResult result = new CallResult();
		result.callId = applyCallId();
		result.timeOut = lastPulseTime+DEFAULT_TIMEOUT;
		result.callbackMethod = callback;
		if(context!=null){
			result.context = context;
			result.handleTimeOut=handleTimeOut;
			result.context.setTimeOut(false);
		}
		if(Config.STAT_ENABLE){
			result.callerInfo = getCallResultCallerInfo();
			result.listenTime=System.currentTimeMillis();
		}
		callbackMap.put(result.callId, result);
		currentCallbackId = result.callId;
	}
	
	/**
	 * 获取调用堆栈
	 * @return
	 */
	private String getCallResultCallerInfo(){
		return "";
	}
	
	/**
	 * 获取回调id，并重置为0
	 * @return
	 */
	public long getAndResetCallBackId(){
		if(currentCallbackId==0){
			return 0;
		}
		long currCallbackId = currentCallbackId;
		currentCallbackId = 0;
		return currCallbackId;
	}
	
	/**
	 * 每秒心跳 
	 */
	public void tickSec(){
		
		checkCallTimeout();
	}
	
	/**
	 * 检测回调是否超时
	 */
	private void checkCallTimeout(){
		lastPulseTime = System.currentTimeMillis();
		List<CallResult> timeOutCall = new ArrayList<>();
		for(CallResult callResult : callbackMap.values()){
			if(lastPulseTime>=callResult.timeOut){
				timeOutCall.add(callResult);
			}
		}
		for(CallResult callResult : timeOutCall){
			callbackMap.remove(callResult.callId);
			if(callResult.handleTimeOut){
				//执行超时回调
				if(callResult.context!=null){
					callResult.context.markTimeOut();
					try {
						FuncUtil.apply(callResult.callbackMethod, callResult.context, null);
					} catch (Exception ex) {
						log.error("CallBack time out handle ERROR! ex"+ex, ex);
					}
				}
			}
		}
	}
	
	/**
	 * 不注册回调，同步等待结果
	 * 
	 * <pre>
	 * 与waitForResult()配对使用，eg:
	 * xx.listenResultSync()
	 * xx.doSomething()
	 * Result ret = xx.waitForResult();
	 * </pre>
	 */
	public void listenSync(){
		if(waitSyncCallId!=0){
			throw new SysException("当前正在等待同步调用结果，不能嵌套同步调用");
		}
		waitSyncCallId = currentCallbackId = applyCallId();
	}
	
	public static <T> T waitFor(IAction action) {
    	return waitFor(action, 60000);
    }
    public static <T> T waitFor(IAction action, long timeout) {
    	ThreadContext current = ThreadContext.getCurrent();
    	if (current == null) {
    		return Context.waitFor(action, timeout);
    	}
    	current.listenSync();
    	action.action();
		return current.waitFor(timeout);
    }
    
	/**
	 * 阻塞当前线程直到返回结果(业务线程中慎用，如果使用不当会造成处理其他逻辑超时，可能对游戏造成严重后果！！！)
	 * 
	 * <pre>
	 * 必须与listenResultSync()配对使用，否则只会返回null。eg:
	 * xx.listenResultSync()
	 * xx.doSomething()
	 * Result ret = xx.waitForResult();
	 * </pre>
	 * @return
	 */
	public <T> T waitFor(){
		return waitFor(DEFAULT_TIMEOUT);
	}
	
	/**
	 * 阻塞等待远程调用返回的结果，此方法谨慎使用，如果使用不当会造成处理其他逻辑超时，可能对游戏造成严重后果！！！
	 * <pre>
	 * 必须与listenResultSync()配对使用，否则只会返回null。eg:
	 * xx.listenResultSync()
	 * xx.doSomething()
	 * Result ret = xx.waitForResult();
	 * </pre>
	 * @param timeout 超时时间（ms）
	 * @return
	 */
	public <T> T waitFor(long timeout){
		if(waitSyncCallId==0){
			//没有标记同步请求结果
			return null;
		}
		if(timeout<=0){
			throw new IllegalArgumentException("同步调用出错，超时时间必须大于0！timeout="+timeout+",stack="+BeanUtil.getStack());
		}
		CallReturn callRet = null;
		long timeOutTime = System.currentTimeMillis()+timeout;
		try {
			while (callRet == null) {
				// 等待请求返回或超时
				while (!callReturns.isEmpty()) {
					CallReturn ret = callReturns.poll();
					if (waitSyncCallId==ret.id) {
						// 得到结果
						callRet = ret;
						break;
					} else {
						onReturn(ret.id, ret.getReturns());
					}
				}
				if (callRet == null) {
					// 超时
					if (System.currentTimeMillis() > timeOutTime) {
						break;
					}
					// 延迟10毫秒进行下一次检查
					Thread.sleep(10);
				}
			}
		} catch (Throwable e) {
			throw new SysException("同步调用出错！", e);
		}finally{
			waitSyncCallId=0;
		}
		if(callRet==null){
			return null;
		}
		return callRet.getReturns();
	}
	
	/**
	 * 收到远程返回
	 * @param callReturn
	 */
	public void onReturn(long callbackId, Object returns){
		//添加到线程进行处理
		CallResult callResult = callbackMap.remove(callbackId);
		if(callResult==null){
			//丢失，不处理,记录日志
			return;
		}
		//开发阶段统计回调时间
		if(Config.STAT_ENABLE){
			long costTime = System.currentTimeMillis()-callResult.listenTime;
			if(costTime>10){//统计超过10ms的回调
				
			}
		}
		//执行回调
		if(callResult.context==null){
			FuncUtil.apply(callResult.callbackMethod, returns);
		}else{
			FuncUtil.apply(callResult.callbackMethod, callResult.context, returns);
		}		
	}
	
	/**
	 * 收到远程返回
	 * @param callReturn
	 */
	public void onReturn(CallReturn callReturn){
		callReturns.add(callReturn);
		//只处理异步回调，同步回调会在waitForResult中处理
		thread.addAction(() -> {
			// 添加到线程进行处理
			while (!callReturns.isEmpty()) {
				CallReturn ret = callReturns.poll();
				onReturn(ret.id, ret.getReturns());
			}
		});
	}


	/**
	 * 将结果返回给调用方
	 */
/*	public void doReturn(){	
		if(callReturn==null){
			return;
		}
		try {
			callReturn.setReturns(returns);
			if (node.getId() == callReturn.nodeId) {
				// 本地调用
				node.addCallReturn(callReturn);
			} else {
				StreamWriter writer = new StreamWriter();
				try {
					callReturn.writeTo(writer);
				} catch (Exception ex) {
					// 记录日志
					log.error("serialize callReturn failed! call=" + currentCall + ", callReturn=" + callReturn + ",ex=" + ex, ex);
					// 清除回调值
					returns = null;
				}
				byte[] calldata = writer.getBuffer();
				NodeMessage nodeMsg = new NodeMessage();
				nodeMsg.setId(SysMessageIds.rpc_return);
				nodeMsg.setMsgData(calldata);
				node.sendNodeMsg(callReturn.nodeId, nodeMsg);
			}
		} catch (Exception ex) {
			log.error("ThreadContext.doReturns err! ex="+ex,ex);
		} finally {
			currentCall = null;
			returns=null;
			callReturn = null;
		}
	}
*/
	/**
	 * 返回远程调用的结果
	 * @param callReturn
	 * @param returns
	 */
/*	public void doReturn(CallReturn callReturn, Object...params){
		doReturn(callReturn, new Param(params));
	}
	*/
	/**
	 * 返回远程调用的结果
	 * @param callReturn
	 * @param returns
	 */
/*	public void doReturn(CallReturn callReturn, Param returns){
		if(callReturn==null){
			return;
		}
		callReturn.setReturns(returns);
		if (node.getId() == callReturn.nodeId) {
			// 本地调用
			node.addCallReturn(callReturn);
		} else {
			StreamWriter writer = new StreamWriter();
			try {
				callReturn.writeTo(writer);
			} catch (Exception ex) {
				// 记录日志
				log.error("serialize callReturn failed! call=" + currentCall + ", callReturn=" + callReturn + ",ex=" + ex, ex);
				// 清除回调值
				returns = null;
			}
			byte[] calldata = writer.getBuffer();
			NodeMessage nodeMsg = new NodeMessage();
			nodeMsg.setId(SysMessageIds.rpc_return);
			nodeMsg.setMsgData(calldata);
			node.sendNodeMsg(callReturn.nodeId, nodeMsg);
		}
	}
	*/
	/**
	 * 设置返回值,只能在服务内返回
	 * @param returns
	 */
/*	public static void returns(Object returns){
		ThreadContext context = ThreadContext.getCurrent();
		if(context!=null){
			context.returns=returns;			
//			context.doReturn();
		}		
	}
	*/
	/**
	 * 设置返回值,格式如：("paramName1",paramVal1,"paramName2",paramVal2)<br />
	 * 对应回调方法的格式应该为 callback(Param returns, Param context)
	 * @param returns
	 */
/*	public static void returns(Object... returns){
		ThreadContext context = ThreadContext.getCurrent();
		if(context!=null){
			Param param = new Param(returns);
			context.returns=param;			
//			context.doReturn();
		}		
	}*/
}
